package tjs.study.notes.flink.simple;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class OperatorTest {
    public static final Logger LOG = LoggerFactory.getLogger(OperatorTest.class);

    public static void main(String[] args) throws Exception {
        //创建执行流处理的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //全局并行度
        // env.setParallelism(10);

        // 从文件读取数据
        //DataStream<String> dataStream = env.readTextFile("/Users/wangyuhang/Desktop/FlinkTutorial/src/main/resources/sensor.txt");
        // 从固定集合读取
        //DataStream<String> dataStream = env.fromCollection(Arrays.asList("A","B,C","C,D,E","C","D"));
        DataStream<String> dataStream = env.addSource(new SourceFunction<String>() {
            @Override
            public void run(SourceContext<String> sourceContext) throws Exception {
                for (int i = 0; i < 1000; i++) {
                    sourceContext.collect("A" + i);
                }
            }

            @Override
            public void cancel() {

            }
        });

        // 1.map.把string转化成长度输出
        DataStream<Integer> mapStream = dataStream.map(new MapFunction<String, Integer>() {
            @Override
            public Integer map(String value) throws Exception {
                LOG.info(">>>>>>>>map：{}", Thread.currentThread().getId()+"@"+Thread.currentThread().getName());
                return value.length();
            }
        });

        //2、flatmap,按照逗号分隔
        DataStream<String> flatMapStream = dataStream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                LOG.info(">>>>>>>>flatMap：{}", Thread.currentThread().getId()+"@"+Thread.currentThread().getName());
                String[] fields = value.split(",");
                for (String field : fields)
                    out.collect(field);
            }
        });

        //3.filter,筛选sensor_1开头的的id对应的数据
        DataStream<String> filterStream = dataStream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                LOG.info(">>>>>>>>flatMap：{}", Thread.currentThread().getId()+"@"+Thread.currentThread().getName());
                return value.startsWith("B");
            }
        });

        //打印输出
        mapStream.print("map");
        flatMapStream.print("flatMap");
        filterStream.print("filter");

        env.execute();
    }
}